常见难点：并发、显存占用与网络瓶颈
学习目标
- 理解大模型推理在生产环境中面临的主要挑战
- 掌握并发请求处理的优化策略与实现方法
- 学习显存管理与优化的关键技术
- 了解网络瓶颈问题及其解决方案
- 熟悉性能监控与故障排查的方法
大模型推理的三大挑战
在将大语言模型部署到生产环境时，通常面临三大核心挑战：并发处理、显存管理和网络通信。这些挑战直接影响服务的性能、可靠性和成本效益。
挑战一：并发请求处理
当多个用户同时向LLM服务发起请求时，如何高效处理并发请求成为首要挑战。
并发问题的表现
- 排队延迟：请求在队列中等待处理时间过长
- 吞吐量瓶颈：系统无法处理预期的请求量
- 资源竞争：多请求争夺有限的计算资源
- 响应时间不稳定：负载变化导致响应时间波动大
通常，在多实例部署架构中，应用级并发处理策略之上还会配合使用外部负载均衡器（如Nginx、HAProxy或云服务商提供的LB服务），它们负责将用户请求分发到不同的推理服务实例，实现初步的负载分担和高可用。
并发处理策略
1. 请求调度与队列管理
问题：简单的先进先出(FIFO)队列在高负载下性能差
解决方案：
- 优先级队列：基于业务重要性划分请求优先级
- 公平调度算法：确保资源在用户间公平分配
- 批处理动态调整：根据负载自动调整批大小
实现示例：使用vLLM的高级调度器
python
from vllm import AsyncLLMEngine, RequestOutput
from vllm.engine.arg_utils import AsyncEngineArgs
from vllm.sampling_params import SamplingParams
import asyncio
# 配置引擎参数
engine_args = AsyncEngineArgs(
model="deepseek-ai/deepseek-llm-7b-chat",
tensor_parallel_size=2,
max_num_batched_tokens=8192,
gpu_memory_utilization=0.85,
)
# 创建异步引擎
engine = AsyncLLMEngine.from_engine_args(engine_args)
# 自定义请求调度
async def handle_request(request_id, prompt, priority=0):
sampling_params = SamplingParams(
temperature=0.7,
max_tokens=100,
)
# 提交请求，带优先级
result_generator = engine.generate(prompt, sampling_params, request_id, priority=priority)
# 等待并返回结果
final_output = None
async for output in result_generator:
final_output = output
return final_output
# 并发处理多请求
async def process_batch(requests):
# 处理一批请求，根据优先级分配资源
tasks = [
handle_request(req["id"], req["prompt"], req.get("priority", 0))
for req in requests
]
return await asyncio.gather(*tasks)2. 连续批处理(Continuous Batching)
问题：传统批处理中，长序列会拖慢整个批次
解决方案：
- 动态批处理：边生成边接收新请求
- 预填充缓存：新请求快速进入生成阶段
- 请求交错：不同阶段的请求交错处理
实现示例：vLLM的连续批处理
python
# vLLM自动实现了连续批处理
# 以下是配置参数以优化连续批处理性能
from vllm import LLM
import time
# 创建引擎时配置
llm = LLM(
model="deepseek-ai/deepseek-llm-7b-chat",
# 最大批处理token数（影响吞吐量和显存使用）
max_num_batched_tokens=16384,
# 控制何时启动推理（取值通常0.1-0.5之间）
inference_batch_size_ratio=0.3,
# 批处理器刷新频率，影响响应延迟
scheduler_delay_ms=100,
)
# 模拟连续请求
prompts = [f"Explain the concept of {topic}" for topic in topics]
for i, prompt in enumerate(prompts):
# 异步提交请求
llm.generate_async(prompt, request_id=i)
# 模拟用户请求间隔
time.sleep(0.1)
# 获取所有结果
results = llm.get_all_results()3. 服务水平保证(SLA)策略
问题：不同用户/请求需要差异化服务质量
解决方案：
- 请求分级：按用户级别或请求类型分级
- 资源隔离：为不同级别分配独立资源池
- 超时与降级：长时间运行的请求自动降级或超时
- 自适应容量规划：根据SLA要求动态调整资源
实现示例：使用FastAPI和vLLM实现SLA分级
python
from fastapi import FastAPI, BackgroundTasks, HTTPException, Header, Depends
from vllm import LLM, SamplingParams
import time
import asyncio
app = FastAPI()
llm = LLM("deepseek-ai/deepseek-llm-7b-chat")
# SLA等级定义
SLA_LEVELS = {
"premium": {"timeout": 30, "priority": 10, "max_tokens": 2048},
"standard": {"timeout": 60, "priority": 5, "max_tokens": 1024},
"basic": {"timeout": 120, "priority": 0, "max_tokens": 512},
}
# 请求处理与SLA控制
@app.post("/generate")
async def generate_text(
request: dict,
background_tasks: BackgroundTasks,
x_api_key: str = Header(None),
):
# 获取用户级别
user_level = get_user_level(x_api_key)
sla = SLA_LEVELS.get(user_level, SLA_LEVELS["basic"])
# 应用SLA限制
request["max_tokens"] = min(request.get("max_tokens", 512), sla["max_tokens"])
# 创建带超时的任务
try:
result = await asyncio.wait_for(
process_request(request, priority=sla["priority"]),
timeout=sla["timeout"]
)
return result
except asyncio.TimeoutError:
# 超时处理
background_tasks.add_task(cancel_request, request["id"])
raise HTTPException(status_code=408, detail="Request timed out")
async def process_request(request, priority=0):
# 实际请求处理逻辑
sampling_params = SamplingParams(
temperature=request.get("temperature", 0.7),
max_tokens=request.get("max_tokens", 512),
)
outputs = llm.generate(request["prompt"], sampling_params)
return {"result": outputs[0].outputs[0].text}
def get_user_level(api_key):
# 从API密钥验证用户级别
# 实际实现应连接到用户数据库或认证服务
user_levels = {
"key1": "premium",
"key2": "standard",
}
return user_levels.get(api_key, "basic")
async def cancel_request(request_id):
# 实现请求取消逻辑，例如通知引擎终止特定请求
# 这在vLLM等框架中可能需要特定的API支持
print(f"Attempting to cancel request {request_id}")
# await engine.abort(request_id) # 假设有这样的方法
pass
#### 4. 请求合并 (Request Batching at the Edge)
**问题**：大量短的、相似的请求可能会频繁调用推理核心，增加调度开销。
**解决方案**：
- 在API网关或代理层，将短时间内到达的多个相似请求合并成一个批次，然后发送给推理引擎。
- 这需要仔细设计合并逻辑，权衡合并带来的延迟增加和吞吐提升。
## 挑战二：显存占用与管理
大语言模型推理过程中，显存管理是影响性能和并发能力的关键因素。
### 显存占用的主要来源
1. **模型权重**：模型参数占用的基础显存
2. **KV缓存**：存储注意力计算中间结果的缓存
3. **激活值**：前向传播中的中间计算结果
4. **优化器状态**：如果进行微调，优化器状态占用额外显存
5. **批处理缓冲区**：处理并发请求的输入/输出缓冲区
### 显存优化策略
#### 1. KV缓存优化
**问题**：KV缓存在长上下文或多请求时迅速膨胀
**解决方案**：
- **分页管理**：PagedAttention/Blocked KV Cache技术
- **缓存裁剪**：移除不重要的历史KV缓存
- **缓存量化**：对KV缓存应用量化技术(INT8/FP8)
- **滑动窗口注意力**：仅保留固定窗口的KV缓存
此外，模型架构本身的改进，如Multi-Query Attention (MQA) 和 Grouped-Query Attention (GQA)，通过让多个查询头共享同一份键（Key）和值（Value）的投影，可以直接减少KV缓存的大小，从而降低显存占用和带宽需求。
对于极长的上下文，还可以考虑将部分不常用的KV缓存卸载（Offload）到CPU内存甚至NVMe固态硬盘，但这会带来显著的访问延迟增加，需要在特定场景下权衡。
**实现示例**：vLLM的PagedAttention配置
```python
# vLLM的PagedAttention配置
from vllm import LLM
llm = LLM(
model="deepseek-ai/deepseek-llm-7b-chat",
# 控制块大小，影响内存利用率和碎片化
# 通常8-128之间，较小值降低碎片但增加管理开销
block_size=16,
# 预分配的GPU显存比例
gpu_memory_utilization=0.85,
# 禁用KV缓存预分配以节省显存（但可能影响性能）
disable_kv_cache_preallocation=True
)2. 模型权重优化
问题：完整精度模型权重占用大量显存
解决方案：
- 模型量化：4/8位量化降低权重占用
- 权重共享：多实例间共享只读权重
- 分层加载：按需加载模型层
- 权重剪枝：去除不重要的权重
模型剪枝（Pruning）和知识蒸馏（Knowledge Distillation）虽然主要在模型训练阶段应用，但其目标是产出更小、更高效的模型，这些模型在推理部署时能直接降低显存占用。要注意的是，这些技术可能会带来一定的精度损失，需要仔细评估。
实现示例：使用bitsandbytes进行权重量化
python
from transformers import AutoModelForCausalLM, AutoTokenizer
import bitsandbytes as bnb
# 4位量化加载模型
model = AutoModelForCausalLM.from_pretrained(
"deepseek-ai/deepseek-llm-7b-chat",
device_map="auto",
load_in_4bit=True,
quantization_config={
"bnb_4bit_compute_dtype": "float16",
"bnb_4bit_use_double_quant": True,
"bnb_4bit_quant_type": "nf4"
}
)
# 检查内存占用
print(f"Model memory footprint: {model.get_memory_footprint() / 1e9:.2f} GB")3. 动态显存管理
问题：固定显存分配无法适应变化的负载
解决方案：
- 显存池化：统一管理和复用显存
- 激进释放：计算完成立即释放临时缓冲区
- 跨设备卸载：将不活跃数据暂存到CPU内存
- 请求级内存预算：根据优先级分配内存配额
诊断显存占用问题时，可以使用专门的显存分析工具，例如 pytorch-memlab (针对PyTorch) 或NVIDIA提供的Nsight系列工具，它们可以帮助定位显存瓶颈和泄漏。
实现示例：DeepSpeed-Inference的ZeRO-Inference
python
import deepspeed
from transformers import AutoModelForCausalLM, AutoTokenizer
# 加载模型
model = AutoModelForCausalLM.from_pretrained("deepseek-ai/deepseek-llm-7b-chat")
tokenizer = AutoTokenizer.from_pretrained("deepseek-ai/deepseek-llm-7b-chat")
# 配置DeepSpeed-Inference
ds_config = {
"tensor_parallel": {
"tp_size": 2
},
"dtype": "fp16",
"injection_policy": {
"attention": {
"mode": "original"
}
},
"replace_method": "auto",
"enable_cuda_graph": False,
"zero_inference": True # 启用ZeRO-Inference
}
# 初始化DeepSpeed引擎
ds_engine = deepspeed.init_inference(
model=model,
config=ds_config,
replace_with_kernel_inject=True
)
# 使用引擎进行推理
input_ids = tokenizer("DeepSeek is", return_tensors="pt").input_ids.cuda()
output = ds_engine.generate(input_ids, max_length=50)挑战三：网络瓶颈
分布式推理和高并发场景下，网络通信可能成为性能瓶颈。
常见网络挑战
- 设备间通信：多GPU/多节点间的数据传输
- 客户端-服务器延迟：请求和响应的网络延迟
- 序列化开销：数据格式转换和序列化的成本
- 带宽受限：网络带宽限制吞吐量
- 网络抖动：不稳定的网络导致性能波动
网络优化策略
1. 设备间通信优化
问题：分布式推理中节点间通信成为瓶颈
解决方案：
- 高速互连：使用NVLink, NVSwitch, InfiniBand等
- 通信算法优化：优化集合通信操作，例如选择最优的AllReduce算法（如Ring AllReduce, Tree AllReduce或针对特定拓扑优化的算法）
- 拓扑感知调度：考虑物理拓扑分配任务
- 通信与计算重叠：并行执行通信和计算
实现示例：针对NCCL的优化配置
python
# 设置NCCL优化选项
import os
# 配置环境变量
os.environ["NCCL_DEBUG"] = "INFO" # 启用调试信息
os.environ["NCCL_IB_DISABLE"] = "0" # 启用InfiniBand
os.environ["NCCL_IB_GID_INDEX"] = "3" # 特定GID索引配置
os.environ["NCCL_IB_HCA"] = "mlx5_0:1,mlx5_1:1" # HCA设备指定
os.environ["NCCL_SOCKET_IFNAME"] = "eth0" # 网络接口指定
# 当使用vLLM等分布式框架时，这些环境变量会自动影响通信性能
# 实际配置应根据具体硬件环境调整2. 流式响应优化
问题：等待完整响应导致首次响应延迟高
解决方案：
- 流式生成：生成一个token立即返回，显著改善交互式应用的感知延迟和用户体验。
- 块级响应：小批量token一起返回
- WebSocket/SSE：使用长连接流式传输
- 增量JSON格式：优化流式JSON数据结构
实现示例：使用FastAPI和SSE实现流式响应
python
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from vllm import LLM, SamplingParams
import json
import asyncio
app = FastAPI()
llm = LLM("deepseek-ai/deepseek-llm-7b-chat")
@app.post("/stream")
async def stream_response(request: Request):
data = await request.json()
prompt = data["prompt"]
# 创建一个异步生成器来流式返回结果
async def generate_stream():
# 设置流式生成参数
sampling_params = SamplingParams(
temperature=0.7,
max_tokens=100,
# 启用流式输出
stream=True
)
# 初始化结果状态
response_id = "resp_" + str(int(time.time()))
chunk_id = 0
# 开始生成并流式返回
outputs = llm.generate(prompt, sampling_params)
for request_output in outputs:
for output in request_output.outputs:
# 仅发送新生成的token
if chunk_id > 0:
new_text = output.text[len(previous_text):]
else:
new_text = output.text
previous_text = output.text
# 创建SSE格式的响应块
chunk = {
"id": f"{response_id}-{chunk_id}",
"object": "text_completion.chunk",
"created": int(time.time()),
"model": "deepseek-7b",
"choices": [
{
"text": new_text,
"index": 0,
"finish_reason": output.finish_reason if output.finished else None
}
]
}
# 发送数据块
yield f"data: {json.dumps(chunk)}\n\n"
# 如果生成完成，发送结束标记
if output.finished:
yield "data: [DONE]\n\n"
break
chunk_id += 1
# 人为添加短暂延迟模拟生成时间，实际部署时移除
await asyncio.sleep(0.02)
# 返回流式响应
return StreamingResponse(
generate_stream(),
media_type="text/event-stream"
)3. 序列化优化
问题：复杂数据结构序列化/反序列化开销大
解决方案：
- 轻量级格式：使用MessagePack等轻量级格式
- 二进制协议：gRPC或自定义二进制协议
- 零拷贝技术：减少内存拷贝
- 增量序列化：只序列化变化的部分
在设计API时，还需考虑请求和响应体结构的 schema 演进与版本控制，以确保客户端和服务器之间的兼容性。
实现示例：使用MessagePack优化序列化
python
from fastapi import FastAPI
import msgpack
from vllm import LLM, SamplingParams
app = FastAPI()
llm = LLM("deepseek-ai/deepseek-llm-7b-chat")
# 配置MessagePack响应
@app.post("/generate/msgpack", response_class=msgpack_response)
async def generate_with_msgpack(request: bytes):
# 反序列化请求
data = msgpack.unpackb(request)
prompt = data["prompt"]
# 生成文本
sampling_params = SamplingParams(
temperature=data.get("temperature", 0.7),
max_tokens=data.get("max_tokens", 100),
)
outputs = llm.generate(prompt, sampling_params)
result = outputs[0].outputs[0].text
# 返回MessagePack格式
return msgpack.packb({
"text": result,
"usage": {
"prompt_tokens": outputs[0].prompt_token_ids.shape[0],
"completion_tokens": len(outputs[0].outputs[0].token_ids),
"total_tokens": outputs[0].prompt_token_ids.shape[0] + len(outputs[0].outputs[0].token_ids)
}
})
# 定义MessagePack响应类
class msgpack_response(PlainTextResponse):
media_type = "application/x-msgpack"
def render(self, content):
return content # 已经是打包好的字节性能监控与故障排查
高效的监控与故障排查是大模型推理系统稳定运行的保障。
关键监控指标
延迟指标
- 首个token延迟(Time to First Token, TTFT)
- 每token生成速度(Tokens per Second, TPS)
- 请求排队延迟
吞吐量指标
- 每秒处理请求数(Requests per Second, RPS)
- 每秒生成token数(Tokens per Second, TPS)
- 批处理利用率
资源使用指标
- GPU利用率
- GPU显存使用量
- CUDA内核执行时间
- CPU利用率
- 系统内存使用
请求指标
- 并发请求数
- 请求队列长度
- 请求成功/失败率
- 请求超时率
监控系统实现
python
# 使用Prometheus和FastAPI实现监控
from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
import prometheus_client as prom
app = FastAPI()
# 创建自定义指标
LATENCY_TTFT = prom.Histogram(
"llm_time_to_first_token_seconds",
"Time to first token in seconds",
buckets=(0.05, 0.1, 0.25, 0.5, 0.75, 1.0, 2.0, 5.0, 10.0)
)
TOKENS_PER_SEC = prom.Histogram(
"llm_tokens_per_second",
"Tokens generated per second",
buckets=(1, 5, 10, 20, 50, 100)
)
GPU_MEMORY_USAGE = prom.Gauge(
"llm_gpu_memory_used_bytes",
"GPU memory used in bytes",
["device"]
)
QUEUE_LENGTH = prom.Gauge(
"llm_request_queue_length",
"Number of requests in queue"
)
# 初始化Prometheus FastAPI监控
Instrumentator().instrument(app).expose(app)
# 请求处理函数中更新指标
@app.post("/generate")
async def generate_text(request: dict):
# 记录请求开始时间
start_time = time.time()
# 更新队列长度
QUEUE_LENGTH.set(llm.get_queue_length())
# 推理处理
outputs = llm.generate(request["prompt"])
# 记录首token延迟
first_token_time = llm.get_ttft()
LATENCY_TTFT.observe(first_token_time)
# 计算token生成速度
tokens_generated = len(outputs[0].outputs[0].token_ids)
generation_time = time.time() - start_time - first_token_time
if generation_time > 0:
tokens_per_second = tokens_generated / generation_time
TOKENS_PER_SEC.observe(tokens_per_second)
# 更新GPU内存使用
for i, mem in enumerate(get_gpu_memory_usage()):
GPU_MEMORY_USAGE.labels(device=f"cuda:{i}").set(mem)
return {"result": outputs[0].outputs[0].text}
# 辅助函数：获取GPU显存使用
def get_gpu_memory_usage():
try:
import torch
return [torch.cuda.memory_allocated(i) for i in range(torch.cuda.device_count())]
except:
return [0]常见问题排查指南
OOM (Out of Memory) 错误
- 检查批处理大小和并发请求数
- 考虑降低精度或使用量化模型
- 增加模型分片或张量并行度
- 排查KV缓存泄漏
响应延迟高
- 分析首token延迟与生成速度分别是否有问题
- 检查GPU利用率，是否有计算瓶颈
- 排查批处理效率，是否有资源浪费
- 检查网络延迟是否是瓶颈
吞吐量不足
- 优化批处理策略，提高GPU利用率
- 检查是否有序列化/反序列化瓶颈
- 评估是否需要扩展更多资源
- 实现请求合并和缓存机制
内存泄漏
- 监控长时间运行下的内存增长
- 检查KV缓存释放是否正确
- 排查Python引用循环导致的泄漏
- 实现定期重置机制
在复杂的微服务架构中，分布式追踪（Distributed Tracing）系统（如OpenTelemetry、Jaeger、Zipkin）对于理解一个请求在不同服务间的完整调用链、定位延迟瓶颈至关重要。通过在请求的生命周期中传播上下文信息，可以清晰地看到每个环节的耗时和依赖关系。
对于新模型、优化策略或推理框架版本的上线，引入A/B测试框架可以帮助在线对比不同方案的实际性能和业务影响，从而做出数据驱动的决策，降低部署风险。
案例研究：企业级DeepSeek部署全面优化
以下是一个综合案例，展示企业级DeepSeek部署如何解决上述所有挑战：
场景需求
- 部署DeepSeek-Chat模型服务于企业内部知识库问答
- 支持300+并发用户，峰值并发请求50+
- 服务质量要求：95%请求首token响应<500ms
- 硬件资源：4×A100-80GB GPU
解决方案架构
一个典型的企业级部署架构可能如下：
mermaid
graph TD
User[用户] --> LB[负载均衡器]
LB --> GW[API网关/服务网格]
GW --> InferenceService1[推理服务实例1]
GW --> InferenceService2[推理服务实例2]
GW --> InferenceServiceN[...]
InferenceService1 --> GPUCluster[GPU集群: vLLM + DeepSeek]
InferenceService2 --> GPUCluster
InferenceServiceN --> GPUCluster
subgraph "监控与告警"
Prometheus[Prometheus]
Grafana[Grafana]
AlertManager[AlertManager]
end
GW --> Prometheus
InferenceService1 --> Prometheus
InferenceService2 --> Prometheus
InferenceServiceN --> Prometheus接入层 (Load Balancer & API Gateway)
- 负载均衡器 (LB): 将外部请求分发到多个API网关实例或直接到推理服务实例。
- API网关 (GW): (例如 Kong, Traefik, 或者云服务商的API Gateway) 负责认证、授权、速率限制、请求路由、基本的请求转换和聚合、日志记录等。它可以保护后端推理服务，并提供统一的API入口。
计算层
- vLLM部署量化DeepSeek模型
- 2×2 张量并行配置跨4张GPU
- PagedAttention高效内存管理
- 连续批处理提高并发处理能力
服务层
- FastAPI提供REST和WebSocket双重接口
- 流式响应机制
- 基于优先级的请求调度
- 多级缓存（结果缓存、KV缓存）
监控层
- Prometheus采集性能指标
- Grafana可视化监控面板
- 自定义告警规则
- 性能日志分析
关键实现代码
python
# 综合解决方案实现
import os
import time
import asyncio
import json
from fastapi import FastAPI, WebSocket, Request, BackgroundTasks, HTTPException, Depends
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
from prometheus_fastapi_instrumentator import Instrumentator
import prometheus_client as prom
from vllm import AsyncLLMEngine, AsyncEngineArgs, SamplingParams, RequestOutput
# 优化GPU通信
os.environ["NCCL_P2P_DISABLE"] = "0"
os.environ["NCCL_IB_DISABLE"] = "0"
# 创建应用
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 监控指标
LATENCY_TTFT = prom.Histogram(
"llm_time_to_first_token_seconds",
"Time to first token in seconds",
buckets=(0.05, 0.1, 0.25, 0.5, 0.75, 1.0, 2.0, 5.0, 10.0)
)
REQ_IN_PROCESS = prom.Gauge(
"llm_requests_in_process",
"Number of requests being processed"
)
Instrumentator().instrument(app).expose(app)
# 请求模型
class GenerationRequest(BaseModel):
prompt: str
max_tokens: int = 512
temperature: float = 0.7
stream: bool = False
priority: int = 0
# 结果缓存
response_cache = {}
# 初始化引擎
async def init_engine():
args = AsyncEngineArgs(
model="deepseek-ai/deepseek-llm-7b-chat",
tensor_parallel_size=2, # 2×2 张量并行
quantization="awq", # 使用AWQ量化
max_num_batched_tokens=16384,
gpu_memory_utilization=0.9,
disable_kv_cache_preallocation=True,
)
return await AsyncLLMEngine.from_engine_args(args)
# 对引擎初始化进行异步处理
@app.on_event("startup")
async def startup():
global engine
engine = await init_engine()
# 同步接口（返回完整结果）
@app.post("/generate")
async def generate_text(request: GenerationRequest):
# 检查缓存
cache_key = f"{request.prompt}_{request.max_tokens}_{request.temperature}"
if cache_key in response_cache:
return response_cache[cache_key]
# 记录开始时间
start_time = time.time()
REQ_IN_PROCESS.inc()
try:
# 设置采样参数
sampling_params = SamplingParams(
temperature=request.temperature,
max_tokens=request.max_tokens,
)
# 执行推理
result_generator = engine.generate(request.prompt, sampling_params, request_id=str(time.time()), priority=request.priority)
# 获取结果
final_output = None
first_token_received = False
async for output in result_generator:
if not first_token_received:
first_token_time = time.time() - start_time
LATENCY_TTFT.observe(first_token_time)
first_token_received = True
final_output = output
# 构建响应
response = {
"text": final_output.outputs[0].text,
"usage": {
"prompt_tokens": len(final_output.prompt_token_ids),
"completion_tokens": len(final_output.outputs[0].token_ids),
"total_tokens": len(final_output.prompt_token_ids) + len(final_output.outputs[0].token_ids),
},
"metrics": {
"time_to_first_token": first_token_time,
"total_time": time.time() - start_time,
}
}
# 存入缓存
response_cache[cache_key] = response
return response
finally:
REQ_IN_PROCESS.dec()
# 流式接口（SSE方式）
@app.post("/stream")
async def stream_generation(request: GenerationRequest):
if not request.stream:
return await generate_text(request)
start_time = time.time()
REQ_IN_PROCESS.inc()
first_token_received = False
async def generate_stream():
nonlocal first_token_received
try:
sampling_params = SamplingParams(
temperature=request.temperature,
max_tokens=request.max_tokens,
)
result_generator = engine.generate(request.prompt, sampling_params, request_id=str(time.time()), priority=request.priority)
response_id = f"resp_{int(time.time())}"
async for output in result_generator:
if not first_token_received:
first_token_time = time.time() - start_time
LATENCY_TTFT.observe(first_token_time)
first_token_received = True
choices = [{
"text": output.outputs[0].text,
"index": 0,
"finish_reason": output.outputs[0].finish_reason if output.outputs[0].finished else None,
}]
chunk = {
"id": response_id,
"object": "text_completion.chunk",
"created": int(time.time()),
"model": "deepseek-7b-chat",
"choices": choices,
}
yield f"data: {json.dumps(chunk)}\n\n"
if output.outputs[0].finished:
yield "data: [DONE]\n\n"
finally:
REQ_IN_PROCESS.dec()
return StreamingResponse(
generate_stream(),
media_type="text/event-stream",
)
# WebSocket接口
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
# 接收请求
data = await websocket.receive_json()
request = GenerationRequest(**data)
start_time = time.time()
REQ_IN_PROCESS.inc()
try:
sampling_params = SamplingParams(
temperature=request.temperature,
max_tokens=request.max_tokens,
)
result_generator = engine.generate(request.prompt, sampling_params, request_id=str(time.time()), priority=request.priority)
first_token_received = False
async for output in result_generator:
if not first_token_received:
first_token_time = time.time() - start_time
LATENCY_TTFT.observe(first_token_time)
first_token_received = True
# 发送增量输出
await websocket.send_json({
"text": output.outputs[0].text,
"finished": output.outputs[0].finished,
"time_to_first_token": first_token_time if first_token_received else None,
})
if output.outputs[0].finished:
break
finally:
REQ_IN_PROCESS.dec()
except Exception as e:
# 处理连接断开等异常
print(f"WebSocket error: {e}")性能结果
上述优化后的系统性能表现：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| 首Token延迟(P95) | 850ms | 290ms | 65.9% |
| 吞吐量(tokens/s) | 28 | 95 | 239.3% |
| 并发请求支持 | 20 | 60 | 200% |
| GPU显存使用效率 | 45% | 85% | 88.9% |
| 请求超时率 | 15% | <1% | >93.3% |
小结
在大模型推理的生产部署中，并发请求处理、显存管理和网络瓶颈是三大核心挑战。通过采用连续批处理、PagedAttention等先进技术，结合精细的请求调度和资源管理策略，可以显著提升系统性能和稳定性。
合理的监控和故障排查体系也是确保服务可靠运行的关键。通过收集和分析关键性能指标，可以及时发现潜在问题并进行优化。
在实际部署中，应根据具体场景和需求，综合应用本节介绍的各种优化策略，找到性能、资源利用和服务质量之间的最佳平衡点。
在下一章中，我们将探讨如何使用Gradio和Streamlit等工具构建简单而强大的LLM应用界面，进一步提升用户体验。